package com.atguigu.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.utils.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.*;

/**
 * ClassName: DwdTableProcessFunction
 * Package: com.atguigu.func
 * Description:
 *
 * @Author Lovxy
 * @Create 2023/5/16 17:46
 * @Version 1.0
 */
public class DwdTableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {

    //定义一个HashMap用于存放预加载的配置信息
    private HashMap<String, TableProcess> configMap;

    private MapStateDescriptor<String, TableProcess> tableProcessStateDes;

    public DwdTableProcessFunction(MapStateDescriptor<String, TableProcess> tableProcessStateDes) {
        this.tableProcessStateDes = tableProcessStateDes;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        configMap = new HashMap<>();
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/edu_config", "root", "000000");
        List<TableProcess> tableProcessList = JdbcUtil.queryList(connection,
                "select * from table_process where sink_type='dwd'",
                TableProcess.class,
                true);

        //将预读取的配置信息存放进Map中
        for (TableProcess tableProcess : tableProcessList) {
            String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
            configMap.put(key, tableProcess);
        }
        connection.close();

    }

    //  value:{"database":"gmall","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":159521118579
//9847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204",
//        "comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext context, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(tableProcessStateDes);
        String key=value.getString("table")+"-"+value.getString("type");
        TableProcess tableProcess = broadcastState.get(key);
        TableProcess tableProcess1 = configMap.get(key);
        if (tableProcess !=null || tableProcess1!=null){
            if (tableProcess == null){
                tableProcess = tableProcess1;
            }
            JSONObject data = value.getJSONObject("data");
            fliterColumns(data,tableProcess.getSinkColumns());
            //补充SinkTable字段并写出数据,未来的主题名
            value.put("sink_table", tableProcess.getSinkTable());
            //输出数据
            out.collect(value);
        } else {
            System.out.println("该组合Key：" + key + "不存在！");
        }

    }

    private void fliterColumns(JSONObject data, String sinkColumns) {
        String[] column = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(column);
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnList.contains(next.getKey()));

    }

    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context context, Collector<JSONObject> out) throws Exception {

        //1.获取广播状态数据
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(tableProcessStateDes);
        //2.解析数据为TableProcess对象
        //如果为删除操作，则删除状态以及map中的配置信息
        JSONObject jsonObject = JSONObject.parseObject(value);
        if ("d".equals(jsonObject.getString("op"))){
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("before"), TableProcess.class);
            String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
            configMap.remove(key);
            broadcastState.remove(key);
        }else {
            TableProcess tableProcess = JSONObject.parseObject(jsonObject.getString("after"), TableProcess.class);
            if ("dwd".equals(tableProcess.getSinkType())){
                String key = tableProcess.getSourceTable() + "-" + tableProcess.getSourceType();
                broadcastState.put(key,tableProcess);
            }

        }

    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
